#pragma once

#include <iostream>
#include <vector>
#include <string>
#include <algorithm>
#include <functional>
#include <unordered_map>
#include <memory>
#include <mutex>
#include <thread>
#include <condition_variable>
#include <stdint.h>
#include <cassert>
#include <cstring>
#include <sys/socket.h>
#include <sys/types.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <signal.h>
#include <sys/epoll.h>
#include <sys/eventfd.h>
#include <sys/timerfd.h>

#include "log.hpp"

#define BUFFER_DEFAULT_SIZE 1024

// 缓冲区类
class Buffer
{
private:
    std::vector<char> _buffer; // 存储数据的数组
    uint64_t _reader_idx;      // 读偏移
    uint64_t _writer_idx;      // 写偏移
public:
    char *Begin() const { return (char *)&*_buffer.begin(); }
    // 获取当前写入位置
    char *WritePosition() const { return Begin() + _writer_idx; }
    // 获取当前读取位置
    char *ReadPosition() const { return Begin() + _reader_idx; }
    // 获取末尾空闲空间 --- 也就是写入位置到数组末尾的长度
    uint64_t TailIdleSize() const { return _buffer.size() - _writer_idx; }
    // 获取起始区域的空闲空间 --- 也就是数组读取位置的下标大小
    uint64_t HeadIdleSize() const { return _reader_idx; }
    // 获取可读数据大小 --- 写偏移 - 读偏移
    uint64_t ReadAbleSize() const { return _writer_idx - _reader_idx; }
    // 将读偏移向后移动
    void MoveReaderOffset(uint64_t len)
    {
        assert(len <= ReadAbleSize());
        _reader_idx += len;
    }
    // 将写偏移向后移动
    void MoveWriterOffset(uint64_t len)
    {
        assert(len <= TailIdleSize());
        _writer_idx += len;
    }
    // 确保写入空间足够
    void EnsureWriteSpace(uint64_t len)
    {
        // 如果末尾位置足够，则直接返回
        if (len <= TailIdleSize())
            return;
        // 如果总空闲空间足够，则移动可读数据
        if (len <= HeadIdleSize() + TailIdleSize())
        {
            uint64_t rsz = ReadAbleSize();
            std::copy(ReadPosition(), ReadPosition() + rsz, Begin());
            _reader_idx = 0;
            _writer_idx = rsz;
        }
        else
        {
            // 如果空闲空间不够就直接扩容
            _buffer.resize(_writer_idx + len);
        }
    }
    // 写入
    void Write(const void *data, uint64_t len)
    {
        if (len == 0)
            return;
        EnsureWriteSpace(len);
        const char *d = (const char *)data;
        std::copy(d, d + len, WritePosition());
    }
    void WriteString(const std::string &data)
    {
        Write(data.c_str(), data.size());
    }
    void WriteBuffer(const Buffer &other_buf)
    {
        Write(other_buf.ReadPosition(), other_buf.ReadAbleSize());
    }
    // 读取
    void Read(void *buf, uint64_t len)
    {
        assert(len <= ReadAbleSize());
        std::copy(ReadPosition(), ReadPosition() + len, (char *)buf);
    }
    std::string ReadAsString(uint64_t len)
    {
        assert(len <= ReadAbleSize());
        std::string str;
        str.resize(len);
        Read(&str[0], len);
        return str;
    }
    // 查找换行符
    char *FindCRLF()
    {
        void *res = memchr(ReadPosition(), '\n', ReadAbleSize());
        return (char *)res;
    }
    std::string GetLine()
    {
        char *pos = FindCRLF();
        if (pos == nullptr)
        {
            return "";
        }
        return ReadAsString(pos - ReadPosition() + 1);
    }

public:
    Buffer() : _buffer(BUFFER_DEFAULT_SIZE), _reader_idx(0), _writer_idx(0) {}
    void WriteAndPush(const void *data, uint64_t len)
    {
        Write(data, len);
        MoveWriterOffset(len);
    }
    void WriteStringAndPush(const std::string &data)
    {
        WriteString(data);
        MoveWriterOffset(data.size());
    }
    void WriteBufferAndPush(const Buffer &other_buf)
    {
        WriteBuffer(other_buf);
        MoveWriterOffset(other_buf.ReadAbleSize());
    }
    void ReadAndPop(void *buf, uint64_t len)
    {
        Read(buf, len);
        MoveReaderOffset(len);
    }
    std::string ReadAsStringAndPop(uint64_t len)
    {
        std::string res = ReadAsString(len);
        MoveReaderOffset(len);
        return res;
    }
    std::string GetLineAndPop()
    {
        std::string str = GetLine();
        MoveReaderOffset(str.size());
        return str;
    }
    // 清空缓冲区
    void Clear()
    {
        // 将读写偏移置零即可
        _writer_idx = 0;
        _reader_idx = 0;
    }
};

#define MAX_LISTEN 10

// 套接字类
class Socket
{
private:
    int _sockfd;

private:
    // 关闭套接字
    void Close()
    {
        if (_sockfd != -1)
        {
            close(_sockfd);
            _sockfd = -1;
        }
    }

public:
    Socket() : _sockfd(-1) {}
    Socket(int sockfd) : _sockfd(sockfd) {}
    ~Socket() { Close(); }
    // 返回fd
    int Fd() { return _sockfd; }
    // 创建套接字
    bool Create()
    {
        _sockfd = socket(AF_INET, SOCK_STREAM, 0);
        if (_sockfd < 0)
        {
            lg(Fatal, "Create socket failed!");
            return false;
        }
        return true;
    }
    // 绑定
    bool Bind(const std::string &ip, uint16_t port)
    {
        struct sockaddr_in local;
        local.sin_family = AF_INET;
        local.sin_port = htons(port);
        local.sin_addr.s_addr = inet_addr(ip.c_str());
        socklen_t len = sizeof(local);

        if (bind(_sockfd, (const sockaddr *)&local, len) < 0)
        {
            lg(Fatal, "Bind failed!");
            return false;
        }
        return true;
    }
    // 设置监听
    bool Listen(int backlog = MAX_LISTEN)
    {
        if (listen(_sockfd, backlog) < 0)
        {
            lg(Fatal, "Listen Fail!");
            return false;
        }
        return true;
    }
    // 向服务端发起新连接
    bool Connect(const std::string &ip, uint16_t port)
    {
        struct sockaddr_in server;
        server.sin_family = AF_INET;
        server.sin_port = htons(port);
        server.sin_addr.s_addr = inet_addr(ip.c_str());
        socklen_t len = sizeof(server);

        if (connect(_sockfd, (const sockaddr *)&server, len) < 0)
        {
            lg(Fatal, "connect failed!");
            return false;
        }
        return true;
    }
    // 获取新连接
    int Accept()
    {
        // lg(Debug, "accepting...");
        int sockfd = accept(_sockfd, nullptr, nullptr);
        if (sockfd < 0)
        {
            lg(Warning, "Accept fail!");
            return -1;
        }
        return sockfd;
    }
    // 接收数据
    ssize_t Recv(void *buf, ssize_t len, int flag = 0)
    {
        int n = recv(_sockfd, buf, len, flag);
        if (n <= 0)
        {
            // EAGAIN : 非阻塞情况下缓冲区没有数据
            // EINTR : 被信号打断
            if (errno == EAGAIN || errno == EINTR)
                return 0;
            lg(Warning, "recv fail.");
            return -1;
        }
        // lg(Info, "recv success: \r\n%s", (char *)buf);
        return n;
    }
    ssize_t NonBlockRecv(void *buf, ssize_t len)
    {
        return Recv(buf, len, MSG_DONTWAIT);
    }
    // 发送数据
    ssize_t Send(const void *buf, ssize_t len, int flag = 0)
    {
        int n = send(_sockfd, buf, len, flag);
        if (n < 0)
        {
            if (errno == EAGAIN || errno == EINTR)
                return 0;
            lg(Warning, "send fail!");
            return -1;
        }
        // lg(Info, "send success:%s", (char *)buf);
        return n;
    }
    ssize_t NonBlockSend(void *buf, ssize_t len)
    {
        return Send(buf, len, MSG_DONTWAIT);
    }
    // 创建一个服务端连接
    bool CreateServer(uint16_t port, const std::string &ip = "0.0.0.0", bool block_flag = true)
    {
        if (!Create())
            return false;
        if (block_flag)
            SetNonBlock();
        if (!Bind(ip, port))
            return false;
        if (!Listen())
            return false;
        SetReuseAddr();
        return true;
    }
    // 创建一个客户端连接
    bool CreateClient(uint16_t serverport, const std::string &serverip)
    {
        if (!Create())
            return false;
        if (!Connect(serverip, serverport))
            return false;
        return true;
    }
    // 设置套接字选项,开启地址端口复用
    void SetReuseAddr()
    {
        int opt = 1;
        setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
        opt = 1;
        setsockopt(_sockfd, SOL_SOCKET, SO_REUSEPORT, &opt, sizeof(opt));
    }
    // 设置套接字属性为非阻塞
    void SetNonBlock()
    {
        int fl = fcntl(_sockfd, F_GETFL);
        if (fl < 0)
        {
            perror("fcntl");
            return;
        }
        fcntl(_sockfd, F_SETFL, fl | O_NONBLOCK);
    }
};

class Epoller;
class EventLoop;
// 事件管理类
class Channel
{
private:
    using EventCallback = std::function<void()>;

private:
    int _fd;
    EventLoop *_loop;
    uint32_t _events;        // 关心的事件类型
    uint32_t _revents;       // 当前连接触发的事件类型
    EventCallback _read_cb;  // 可读事件回调
    EventCallback _write_cb; // 可写事件回调
    EventCallback _error_cb; // 错误事件回调
    EventCallback _close_cb; // 连接断开事件被触发的回调
    EventCallback _event_cb; // 任意事件被触发的回调
public:
    Channel(int fd, EventLoop *loop) : _fd(fd), _events(0), _revents(0), _loop(loop) {};
    // 返回当前channel对应的fd
    int Fd() { return _fd; }
    // 返回关心事件
    uint32_t Events() { return _events; }
    void SetRevents(const uint32_t revents) { _revents = revents; }
    // 设置回调方法
    void SetReadCallback(const EventCallback &cb) { _read_cb = cb; }
    void SetWriteCallback(const EventCallback &cb) { _write_cb = cb; }
    void SetErrorCallback(const EventCallback &cb) { _error_cb = cb; }
    void SetCloseCallback(const EventCallback &cb) { _close_cb = cb; }
    void SetEventCallback(const EventCallback &cb) { _event_cb = cb; }
    // 查询当前是否可读/可写
    bool ReadAble() { return (_events & EPOLLIN); }
    bool WriteAble() { return (_events & EPOLLOUT); }
    // 设置读事件/写事件关心
    void EnableRead()
    {
        _events |= EPOLLIN;
        Update();
    }
    void EnableWrite()
    {
        _events |= EPOLLOUT;
        Update();
    }
    // 取消读事件/写事件/所有事件关心
    void DisableRead()
    {
        _events &= ~EPOLLIN;
        Update();
    }
    void DisableWrite()
    {
        _events &= ~EPOLLOUT;
        Update();
    }
    void DisableAll()
    {
        _events = 0;
        Update();
    }
    // 移除对当前channel的关心
    void Remove();
    // 修改关心事件
    void Update();
    // 事件处理，处理被触发的关心事件
    void HandleEvent()
    {
        // 三种读事件事件需要关心，触发后需要调用读回调 : EPOLLIN EPOLLRDHUP(关闭连接，需要处理缓冲区数据) EPOLLPRI(优先数据)
        if ((_revents & EPOLLIN) || (_revents & EPOLLRDHUP) || (_revents & EPOLLPRI))
        {
            if (_read_cb)
                _read_cb();
        }
        // 写事件 EPOLLOUT 错误事件 EPOLLERR 关闭连接的事件 EPOLLHUP
        // 对于可能关闭连接的事件，只执行其中的一个
        if (_revents & EPOLLOUT)
        {
            if (_write_cb)
                _write_cb();
        }
        else if (_revents & EPOLLERR) /*释放连接前执行任意事件触发的回调*/
        {
            if (_error_cb)
                _error_cb();
        }
        else if (_revents & EPOLLHUP)
        {
            if (_close_cb)
                _close_cb();
        }
        // 由于前面的操作不再会直接释放连接，所以可以在后面进行活跃度刷新
        if (_event_cb)
            _event_cb();
    }
};

#define MAX_EPOLLEVENTS_SIZE 1024

// 事件监控模块
class Epoller
{
private:
    int _epfd;
    // 存放epoll_wait返回的关心事件
    struct epoll_event _evs[MAX_EPOLLEVENTS_SIZE];
    // 文件描述符和对应Channel对象指针的映射关系
    std::unordered_map<int, Channel *> _channels;

private:
    void Update(Channel *channel, int op)
    {
        int fd = channel->Fd();
        struct epoll_event ev;
        ev.data.fd = fd;
        ev.events = channel->Events();
        int n = epoll_ctl(_epfd, op, fd, &ev);
        if (n < 0)
            lg(Error, "epoll_ctl fail!");
    }
    bool HasChannel(Channel *channel)
    {
        auto it = _channels.find(channel->Fd());
        if (it == _channels.end())
            return false;
        return true;
    }

public:
    Epoller()
    {
        _epfd = epoll_create(MAX_EPOLLEVENTS_SIZE);
        if (_epfd == -1)
        {
            lg(Warning, "epoll create fail!, strerr: %s", strerror(errno));
            abort();
        }
        lg(Info, "epoll create success! _epfd: %d", _epfd);
    }
    void UpdateEvent(Channel *channel)
    {
        bool hc = HasChannel(channel);
        if (!hc)
        {
            // 不存在则添加
            Update(channel, EPOLL_CTL_ADD);
            _channels[channel->Fd()] = channel;
        }
        else
        {
            Update(channel, EPOLL_CTL_MOD);
        }
    }
    void RemoveEvent(Channel *channel)
    {
        auto it = _channels.find(channel->Fd());
        if (it != _channels.end())
            _channels.erase(it);
        Update(channel, EPOLL_CTL_DEL);
    }
    void Epoll(std::vector<Channel *> *active)
    {
        int n = epoll_wait(_epfd, _evs, MAX_EPOLLEVENTS_SIZE, -1);
        if (n < 0)
        {
            if (errno == EINTR)
                return;
            lg(Fatal, "epoll_wait fail! error : %s", strerror(errno));
            abort();
        }
        for (int i = 0; i < n; ++i)
        {
            auto it = _channels.find(_evs[i].data.fd);
            assert(it != _channels.end());
            it->second->SetRevents(_evs[i].events);
            active->push_back(it->second);
        }
    }
};

class EventLoop;

using Task_Func = std::function<void()>;
using Release_Func = std::function<void()>;

class TimerTask
{
private:
    uint64_t _id;             // 任务的id
    uint32_t _timeout;        // 任务的延时时间
    bool _cancel;             // 表示当前任务是否被取消
    Task_Func _task_cb;       // 执行任务的回调函数
    Release_Func _release_cb; // 释放当前任务在时间轮中的映射
public:
    TimerTask(uint64_t id, uint32_t delay, const Task_Func &tcb)
        : _id(id), _timeout(delay), _task_cb(tcb), _cancel(false)
    {
    }
    ~TimerTask()
    {
        if (!_cancel)
            _task_cb();
        _release_cb();
    }

public:
    void SetRelease(const Release_Func &rcb)
    {
        _release_cb = rcb;
    }

    void SetCancel()
    {
        _cancel = true;
    }

    uint32_t Timeout()
    {
        return _timeout;
    }
};

class TimerWheel
{
private:
    using WeakTask = std::weak_ptr<TimerTask>;
    using TaskPtr = std::shared_ptr<TimerTask>;
    const static int defaultsize = 60;

private:
    int _tick;                                      // 当前的时间指针，走到哪里就释放哪里的所有延时任务节点
    std::vector<std::vector<TaskPtr>> _wheel;       // 存放延时任务定时器的时间轮数组
    std::unordered_map<uint64_t, WeakTask> _timers; // id和对应任务weak_ptr的映射关系

    int _timerfd;
    EventLoop *_loop;
    std::unique_ptr<Channel> _timer_channel;

private:
    void RemoveTimer(uint64_t id)
    {
        auto it = _timers.find(id);
        if (it == _timers.end())
            return;
        _timers.erase(it);
    }

    static int TimerCreate()
    {
        int timerfd = timerfd_create(CLOCK_MONOTONIC, 0);
        if (timerfd < 0)
        {
            lg(Error, "timerfd create fail!");
            abort();
        }
        struct itimerspec itime;
        itime.it_value.tv_sec = 1;
        itime.it_value.tv_nsec = 0; // 第一次超时时间为1s后
        itime.it_interval.tv_sec = 1;
        itime.it_interval.tv_nsec = 0; // 第一次超时后，每隔1s超时一次
        timerfd_settime(timerfd, 0, &itime, nullptr);
        return timerfd;
    }

    int ReadTimerfd()
    {
        uint64_t times;
        int n = read(_timerfd, &times, sizeof(times));
        if (n < 0)
        {
            lg(Error, "read timerfd fail.");
            abort();
        }
        return times;
    }

    void RunTimerTask()
    {
        _tick = (_tick + 1) % _wheel.size();
        _wheel[_tick].clear();
    }

    void OnTime()
    {
        // 每隔一秒超时一次并返回
        int times = ReadTimerfd();
        // 让指针向后走一格
        for (int i = 0; i < times; ++i)
            RunTimerTask();
    }

    void AddTimerInLoop(uint64_t id, uint32_t delay, const Task_Func &tcb)
    {
        TaskPtr tp(new TimerTask(id, delay, tcb));
        tp->SetRelease(std::bind(&TimerWheel::RemoveTimer, this, id));
        int pos = (_tick + delay) % _wheel.size();
        _wheel[pos].push_back(tp);
        _timers[id] = WeakTask(tp);
    }

    void RefreshTimerInLoop(uint64_t id)
    {
        auto it = _timers.find(id);
        // 找不到当前任务的id
        if (it == _timers.end())
            return;
        TaskPtr tp = it->second.lock();
        uint32_t timeout = tp->Timeout();
        // 注意：时间指针是循环增加的
        int pos = (_tick + timeout) % _wheel.size();
        _wheel[pos].push_back(tp);
    }

    void CancelTimerInLoop(uint64_t id)
    {
        auto it = _timers.find(id);
        // 找不到当前任务的id
        if (it == _timers.end())
            return;
        TaskPtr tp = it->second.lock();
        if (tp)
            tp->SetCancel();
    }

public:
    TimerWheel(EventLoop *loop)
        : _tick(0),
          _wheel(defaultsize),
          _timerfd(TimerCreate()),
          _loop(loop),
          _timer_channel(new Channel(_timerfd, _loop))
    {
        _timer_channel->SetReadCallback(std::bind(&TimerWheel::OnTime, this));
        _timer_channel->EnableRead();
    }

    // 对定时任务的操作必须是线程安全的，于是就将这些操作放入一个EventLoop中串行执行
    void AddTimer(uint64_t id, uint32_t delay, const Task_Func &tcb);
    void RefreshTimer(uint64_t id);
    void CancelTimer(uint64_t id);

    bool HasTimer(uint64_t id)
    {
        auto it = _timers.find(id);
        if (it != _timers.end())
            return true;
        return false;
    }
};

class EventLoop
{
private:
    using Functor = std::function<void()>;

private:
    std::thread::id _thread_id;              // 当前对象对应的线程id
    int _eventfd;                            // 事件通知描述符
    std::unique_ptr<Channel> _event_channel; // 事件通知描述符对应的Channel对象指针
    Epoller _epoller;
    std::vector<Functor> _tasks; // 任务队列，保存未完成的任务
    std::mutex _mutex;           // 确保对任务队列进行操作时的线程安全
    TimerWheel _timer_wheel;     // 时间轮

private:
    // 执行所有任务池中的任务
    void RunAllTask()
    {
        std::vector<Functor> tmp;
        {
            std::unique_lock<std::mutex> _lock(_mutex);
            tmp.swap(_tasks);
        }
        for (auto &task : tmp)
            task();
    }
    static int CreateEventFd()
    {
        int efd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
        if (efd < 0)
        {
            lg(Error, "create eventfd fail");
            abort();
        }
        return efd;
    }
    void ReadEventFd()
    {
        uint64_t res = 0;
        int n = read(_eventfd, &res, sizeof(res));
        if (n < 0)
        {
            if (errno == EINTR || errno == EAGAIN)
            {
                return;
            }
            lg(Error, "read event fail");
            abort();
        }
    }
    void WakeupEventFd()
    {
        uint64_t val = 1;
        int n = write(_eventfd, &val, sizeof(val));
        if (n < 0)
        {
            if (errno == EINTR)
            {
                return;
            }
            lg(Error, "write event fail");
            abort();
        }
    }

public:
    EventLoop()
        : _thread_id(std::this_thread::get_id()),
          _eventfd(CreateEventFd()),
          _event_channel(new Channel(_eventfd, this)),
          _timer_wheel(this)
    {
        _event_channel->SetReadCallback(std::bind(&EventLoop::ReadEventFd, this));
        _event_channel->EnableRead();
    }
    // 事件监控  事件处理  任务执行
    void Start()
    {
        while (true)
        {
            std::vector<Channel *> active_channels;
            _epoller.Epoll(&active_channels);

            for (auto &ac : active_channels)
                ac->HandleEvent();

            RunAllTask();
        }
    }
    // 判断任务是否在当前线程中执行
    bool IsInLoop(const Functor &cb)
    {
        return _thread_id == std::this_thread::get_id();
    }
    // 断言操作一定处在当前线程
    void AssertInLoop()
    {
        assert(_thread_id == std::this_thread::get_id());
    }
    // 如果任务在当前线程中则直接执行，不是则放入任务池
    void RunInLoop(const Functor &cb)
    {
        if (IsInLoop(cb))
        {
            cb();
        }
        else
        {
            QueueInLoop(cb);
        }
    }
    // 将任务放入任务池
    void QueueInLoop(const Functor &cb)
    {
        {
            std::unique_lock<std::mutex> _lock(_mutex);
            _tasks.push_back(cb);
        }
        // 唤醒因为没有事件就绪而导致的epoll阻塞
        WakeupEventFd();
    }
    void UpdateEvent(Channel *channel) { _epoller.UpdateEvent(channel); }
    void RemoveEvent(Channel *channel) { _epoller.RemoveEvent(channel); }
    void AddTimer(uint64_t id, uint32_t delay, const Task_Func &tcb) { _timer_wheel.AddTimer(id, delay, tcb); }
    void RefreshTimer(uint64_t id) { _timer_wheel.RefreshTimer(id); }
    void CancelTimer(uint64_t id) { _timer_wheel.CancelTimer(id); }
    bool HasTimer(uint64_t id) { return _timer_wheel.HasTimer(id); }
};

void Channel::Update() { _loop->UpdateEvent(this); }
void Channel::Remove() { _loop->RemoveEvent(this); }

void TimerWheel::AddTimer(uint64_t id, uint32_t delay, const Task_Func &tcb)
{
    _loop->RunInLoop(std::bind(&TimerWheel::AddTimerInLoop, this, id, delay, tcb));
}

void TimerWheel::RefreshTimer(uint64_t id)
{
    _loop->RunInLoop(std::bind(&TimerWheel::RefreshTimerInLoop, this, id));
}

void TimerWheel::CancelTimer(uint64_t id)
{
    _loop->RunInLoop(std::bind(&TimerWheel::CancelTimerInLoop, this, id));
}

class LoopThread
{
private:
    EventLoop *_loop;    // 线程所对应的事件循环对应的指针
    std::thread _thread; // 线程

    std::mutex _mutex;             // 互斥锁
    std::condition_variable _cond; // 条件变量

private:
    // 实例化线程，并且开始运行线程对应EventfdLoop的功能
    void ThreadEntry()
    {
        EventLoop loop;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _loop = &loop;
            _cond.notify_all();
        }
        loop.Start();
    }

public:
    // 创建线程，设定线程的入口函数
    LoopThread() : _loop(nullptr), _thread(&LoopThread::ThreadEntry, this) {}
    // 获取线程对应的EventfdLoop
    EventLoop *GetLoop()
    {
        EventLoop *loop = nullptr;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _cond.wait(lock, [&]
                       { return _loop != nullptr; });
            loop = _loop;
        }
        return loop;
    }
};

class LoopThreadPool
{
private:
    int _thread_count; // 线程数量
    int _next_idx;     // 待分配连接的下一个线程下标

    EventLoop *_base_loop;              // 主线程对应的事件循环
    std::vector<LoopThread *> _threads; // 存放的线程
    std::vector<EventLoop *> _loops;    // 线程对应的事件循环
public:
    LoopThreadPool(EventLoop *bl) : _thread_count(0), _next_idx(-1), _base_loop(bl) {}
    void SetThreadcount(int tc) { _thread_count = tc; }
    // 创建线程池
    void Create()
    {
        _threads.resize(_thread_count);
        _loops.resize(_thread_count);
        for (int i = 0; i < _thread_count; ++i)
        {
            _threads[i] = new LoopThread;
            _loops[i] = _threads[i]->GetLoop();
        }
    }
    EventLoop *NextLoop()
    {
        if (_thread_count == 0)
            return _base_loop;
        _next_idx = (_next_idx + 1) % _thread_count;
        return _loops[_next_idx];
    }
};

class Any
{
private:
    class holder
    {
    public:
        virtual ~holder() {}
        virtual const std::type_info &type() = 0;
        virtual holder *clone() = 0;
    };
    template <class T>
    class placeholder : public holder
    {
    public:
        placeholder(const T &val) : _val(val) {}
        // 获取子类对象保存的数据类型
        virtual const std::type_info &type() { return typeid(T); }
        // 针对当前的对象自身，克隆出一个新的子类对象
        virtual holder *clone() { return new placeholder(_val); }

    public:
        T _val;
    };
    holder *_content;

public:
    Any() : _content(NULL) {}
    template <class T>
    Any(const T &val) : _content(new placeholder<T>(val)) {}
    Any(const Any &other) : _content(other._content ? other._content->clone() : NULL) {}
    ~Any() { delete _content; }

    Any &swap(Any &other)
    {
        std::swap(_content, other._content);
        return *this;
    }

    // 返回子类对象保存的数据的指针
    template <class T>
    T *get()
    {
        // 想要获取的数据类型，必须和保存的数据类型一致
        assert(typeid(T) == _content->type());
        return &((placeholder<T> *)_content)->_val;
    }
    // 赋值运算符的重载函数
    template <class T>
    Any &operator=(const T &val)
    {
        // 为val构造一个临时的通用容器，然后与当前容器自身进行指针交换，临时对象释放的时候，原先保存的数据也就被释放
        Any(val).swap(*this);
        return *this;
    }
    Any &operator=(const Any &other)
    {
        Any(other).swap(*this);
        return *this;
    }
};

enum ConnStatu
{
    DISCONNECTED, // 连接关闭
    CONNECTING,   // 连接建立成功，待处理
    CONNECTED,    // 连接建立完成,设置已完成，可以进行通信
    DISCONNECTING // 待关闭状态
};

class Connection;
using ConnectionPtr = std::shared_ptr<Connection>;
class Connection : public std::enable_shared_from_this<Connection>
{
private:
    using ConnectedCallback = std::function<void(const ConnectionPtr &)>;
    using MessageCallback = std::function<void(const ConnectionPtr &, Buffer *)>;
    using CloseCallback = std::function<void(const ConnectionPtr &)>;
    using AnyEventCallback = std::function<void(const ConnectionPtr &)>;

private:
    uint64_t _conn_id;             // 唯一标识连接的id
    int _sockfd;                   // 连接关联的文件描述符
    bool _enable_inactive_release; // 是否开启非活跃连接释放
    EventLoop *_loop;
    ConnStatu _statu;  // 连接状态
    Socket _socket;    // 套接字管理操作
    Channel _channel;  // 连接的事件管理
    Buffer _inbuffer;  // 输入缓冲区
    Buffer _outbuffer; // 输出缓冲区
    Any _context;      // 请求的接收处理上下文

    /* 这四个回调函数都是由组件使用者设置的 */
    ConnectedCallback _connected_cb;
    MessageCallback _message_cb;
    CloseCallback _close_cb;
    AnyEventCallback _anyevent_cb;
    // 组件内的关闭回调函数
    CloseCallback _server_close_cb;

private:
    /* 五个channel的事件回调函数 */
    // 描述符可读事件触发调用的处理函数
    void HandleRead()
    {
        // 从套接字中读取数据
        char buf[65535];
        ssize_t ret = _socket.NonBlockRecv(buf, 65535);
        if (ret < 0)
        {
            // 出错并不立即关闭连接
            lg(Warning, "recv fail!");
            ShutdownInLoop();
        }
        // 将数据写入缓冲区
        // lg(Debug,"read message: %s", buf);
        _inbuffer.WriteAndPush(buf, ret);
        if (_inbuffer.ReadAbleSize() > 0)
        {
            // 如果从缓冲区中成功写入数据，则交给上层处理
            _message_cb(shared_from_this(), &_inbuffer);
        }
    }
    // 描述符可写事件触发调用的处理函数
    void HandleWrite()
    {
        // 使用套接字发送输出缓冲区中的数据
        int ret = _socket.NonBlockSend(_outbuffer.ReadPosition(), _outbuffer.ReadAbleSize());
        if (ret < 0)
        {
            HandleClose();
        }
        // 由于是对输出缓冲区的数据进行直接输出，因此要手动移动读偏移
        _outbuffer.MoveReaderOffset(ret);
        if (_outbuffer.ReadAbleSize() == 0)
        {
            // 关闭写事件关心
            _channel.DisableWrite();
            // 如果状态为待关闭，则关闭连接
            if (_statu == DISCONNECTING)
                Release();
        }
    }
    // 描述符挂断事件触发调用的处理函数
    void HandleClose()
    {
        // 如果发生错误事件，则先看是否还有数据在输入缓冲区中还没有处理
        if (_inbuffer.ReadAbleSize() > 0)
        {
            _message_cb(shared_from_this(), &_inbuffer);
        }
        // 关闭连接
        Release();
    }
    // 描述符错误事件触发调用的处理函数
    void HandleError()
    {
        HandleClose();
    }
    // 描述符任意事件触发调用的处理函数
    void HandleEvent()
    {
        // 如果设置了非活跃连接释放，则在任意事件触发时进行刷新活跃度
        if (_enable_inactive_release)
            _loop->RefreshTimer(_conn_id);
        if (_anyevent_cb)
            _anyevent_cb(shared_from_this());
    }
    void EstablishedInLoop()
    {
        // 1.修改连接状态 2.启动读事件监控 3.调用回调函数
        assert(_statu == CONNECTING);
        _statu = CONNECTED;
        _channel.EnableRead();
        // lg(Debug,"connection id:%d enableread", Id());
        if (_connected_cb)
            _connected_cb(shared_from_this());
    }
    // 实际的连接释放接口
    void ReleaseInLoop()
    {
        _statu = DISCONNECTED;
        _channel.Remove();
        if (_loop->HasTimer(_conn_id))
            CancelInactiveReleaseInLoop();
        if (_close_cb)
            _close_cb(shared_from_this());
        // 服务器中连接管理信息的释放
        if (_server_close_cb)
            _server_close_cb(shared_from_this());
    }
    void Release()
    {
        _loop->QueueInLoop(std::bind(&Connection::ReleaseInLoop, this));
    }
    // 并非真正的发送数据，而是将数据写入输出缓冲区并打开写事件关心
    void SendInLoop(Buffer buf)
    {
        if (_statu == DISCONNECTED)
            return;
        _outbuffer.WriteBufferAndPush(buf);
        if (!_channel.WriteAble())
            _channel.EnableWrite();
    }
    // 并非真正的连接关闭操作还需要判断是否还存在数据需要处理
    void ShutdownInLoop()
    {
        // 设置连接状态为待关闭
        _statu = DISCONNECTING;
        // 将输入缓冲区的数据全部交给上层
        if (_inbuffer.ReadAbleSize() > 0)
            _message_cb(shared_from_this(), &_inbuffer);
        // 处理后如果输出缓冲区有数据就进行处理
        if (_outbuffer.ReadAbleSize() > 0)
        {
            if (!_channel.WriteAble())
                _channel.EnableWrite();
        }
        // 数据处理完毕，关闭连接
        if (_outbuffer.ReadAbleSize() == 0)
        {
            // 没有数据需要处理
            Release();
        }
    }
    // 启动非活跃连接释放
    void EnableInactiveReleaseInLoop(int sec)
    {
        _enable_inactive_release = true;
        // 当前连接的定时销毁是否存在时间轮，存在则刷新，不存在则添加
        if (_loop->HasTimer(_conn_id))
            _loop->RefreshTimer(_conn_id);
        else
            _loop->AddTimer(_conn_id, sec, std::bind(&Connection::Release, this));
    }
    // 取消非活跃连接释放
    void CancelInactiveReleaseInLoop()
    {
        _enable_inactive_release = false;
        if (_loop->HasTimer(_conn_id))
            _loop->CancelTimer(_conn_id);
    }
    // 切换协议  重置上下文及回调函数
    void UpdateGreadeInLoop(const Any &context,
                            const ConnectedCallback &conncb,
                            const MessageCallback &msgcb,
                            const CloseCallback &closecb,
                            const AnyEventCallback &aetcb)
    {
        _context = context;
        _connected_cb = conncb;
        _message_cb = msgcb;
        _close_cb = closecb;
        _anyevent_cb = aetcb;
    }

public:
    Connection(EventLoop *loop, uint64_t connid, int sockfd) : _conn_id(connid),
                                                               _sockfd(sockfd),
                                                               _enable_inactive_release(false),
                                                               _loop(loop),
                                                               _statu(CONNECTING),
                                                               _socket(_sockfd),
                                                               _channel(_sockfd, _loop)

    {
        _channel.SetReadCallback(std::bind(&Connection::HandleRead, this));
        _channel.SetWriteCallback(std::bind(&Connection::HandleWrite, this));
        _channel.SetErrorCallback(std::bind(&Connection::HandleError, this));
        _channel.SetCloseCallback(std::bind(&Connection::HandleClose, this));
        _channel.SetEventCallback(std::bind(&Connection::HandleEvent, this));
    }
    ~Connection()
    {
        lg(Info, "connection_id:%d closed.", _conn_id);
    }
    int Id() { return _conn_id; }
    int Fd() { return _sockfd; }
    bool IsConnected() { return _statu == CONNECTED; }
    void SetContext(const Any &context) { _context = context; }
    Any *GetContext() { return &_context; }
    void SetConnectedCallback(const ConnectedCallback &cb) { _connected_cb = cb; }
    void SetMessageCallback(const MessageCallback &cb) { _message_cb = cb; }
    void SetCloseCallback(const CloseCallback &cb) { _close_cb = cb; }
    void SetSvrClosedCallback(const ConnectedCallback &cb) { _server_close_cb = cb; }
    void SetAnyEventCallback(const AnyEventCallback &cb) { _anyevent_cb = cb; }
    // 连接建立后设置读事件
    void Established()
    {
        // lg(Debug,"conn:%d use Established()");
        _loop->RunInLoop(std::bind(&Connection::EstablishedInLoop, this));
    }
    // 将数据发送到缓冲区
    void Send(const char *data, size_t len)
    {
        // 输入的数据可能是临时空间，而这个任务不一定会立即执行，可能等到任务执行的时候data已经被释放了
        Buffer buf;
        buf.WriteAndPush(data, len);
        _loop->RunInLoop(std::bind(&Connection::SendInLoop, this, std::move(buf)));
    }
    // 关闭连接，但要先判断数据是否被处理完
    void Shutdown()
    {
        _loop->RunInLoop(std::bind(&Connection::ShutdownInLoop, this));
    }
    // 设置非活跃连接释放
    void EnableInactiveRelease(int sec)
    {
        _loop->RunInLoop(std::bind(&Connection::EnableInactiveReleaseInLoop, this, sec));
    }
    // 取消非活跃连接释放
    void CancelInactiveRelease()
    {
        _loop->RunInLoop(std::bind(&Connection::CancelInactiveReleaseInLoop, this));
    }
    // 切换协议  重置上下文及回调函数
    // 保证协议切换立即执行，防止还有数据使用原协议进行处理
    void UpdateGreade(const Any &context,
                      const ConnectedCallback &conncb,
                      const MessageCallback &msgcb,
                      const CloseCallback &closecb,
                      const AnyEventCallback &aetcb)
    {
        _loop->RunInLoop(std::bind(&Connection::UpdateGreadeInLoop, this, context, conncb, msgcb, closecb, aetcb));
    }
};

class Acceptor
{
private:
    using AcceptCallback = std::function<void(int)>;

private:
    Socket _socket;   // 套接字，用于接受新连接
    EventLoop *_loop; // 用于监控acceptor的事件
    Channel _channel; // 用于管理acceptor的事件

    AcceptCallback _accept_cb; // 处理新连接
private:
    //  获取新连接并调用回调函数处理新连接
    void HandleRead()
    {
        int newfd = _socket.Accept();
        if (newfd < 0)
        {
            lg(Warning, "accept failed");
            return;
        }
        if (_accept_cb)
            _accept_cb(newfd);
    }
    int CreateServer(int port)
    {
        if (!_socket.CreateServer(port))
        {
            lg(Fatal, "create Server failed");
            abort();
        }
        return _socket.Fd();
    }

public:
    Acceptor(EventLoop *lp, int port) : _loop(lp), _channel(CreateServer(port), lp)
    {
        _channel.SetReadCallback(std::bind(&Acceptor::HandleRead, this));
    }
    ~Acceptor() {}
    void SetAcceptCallback(const AcceptCallback &cb) { _accept_cb = cb; }
    void Listen()
    {
        _channel.EnableRead();
    }
};

class TcpServer
{
private:
    using Functor = std::function<void()>;
    using ConnectedCallback = std::function<void(const ConnectionPtr &)>;
    using MessageCallback = std::function<void(const ConnectionPtr &, Buffer *)>;
    using CloseCallback = std::function<void(const ConnectionPtr &)>;
    using AnyEventCallback = std::function<void(const ConnectionPtr &)>;

private:
    int _port;                                          // 端口号
    int _next_id;                                       // 连接编号
    int _timeout;                                       // 超时时间
    bool _enable_inactive_release;                      // 是否启动非活跃连接释放
    EventLoop _base_loop;                               // 主线程对应的事件循环
    Acceptor _acceptor;                                 // 接收连接
    LoopThreadPool _pool;                               // 从属线程池
    std::unordered_map<uint64_t, ConnectionPtr> _conns; // 存放所有接收到的连接

    ConnectedCallback _connected_cb;
    MessageCallback _message_cb;
    CloseCallback _close_cb;
    AnyEventCallback _anyevent_cb;

private:
    void RunAfterInLoop(const Functor &task, int delay)
    {
        ++_next_id;
        _base_loop.AddTimer(_next_id, delay, task);
    }
    void NewConntion(int fd)
    {
        ++_next_id;
        ConnectionPtr conn(new Connection(_pool.NextLoop(), _next_id, fd));
        conn->SetConnectedCallback(_connected_cb);
        conn->SetMessageCallback(_message_cb);
        conn->SetCloseCallback(_close_cb);
        conn->SetSvrClosedCallback(std::bind(&TcpServer::RemoveConntion, this, std::placeholders::_1));
        if (_enable_inactive_release)
            conn->EnableInactiveRelease(_timeout);
        conn->Established();
        _conns[_next_id] = conn;
        lg(Debug, "a new cilent:%d connected.", conn->Id());
    }
    void RemoveConntionInLoop(const ConnectionPtr &conn)
    {
        lg(Info, "connection id:%d destroied.", conn->Id());
        _conns.erase(conn->Id());
    }
    // 从_conns中移除连接信息
    void RemoveConntion(const ConnectionPtr &conn)
    {
        _base_loop.RunInLoop(std::bind(&TcpServer::RemoveConntionInLoop, this, conn));
    }

public:
    TcpServer(int port) : _port(port),
                          _next_id(0),
                          _enable_inactive_release(false),
                          _acceptor(&_base_loop, port),
                          _pool(&_base_loop)
    {
        _acceptor.SetAcceptCallback(std::bind(&TcpServer::NewConntion, this, std::placeholders::_1));
        _acceptor.Listen();
    }
    void SetThreadcount(int count) { _pool.SetThreadcount(count); }
    void SetConnectedCallback(const ConnectedCallback &cb) { _connected_cb = cb; }
    void SetMessageCallback(const MessageCallback &cb) { _message_cb = cb; }
    void SetCloseCallback(const CloseCallback &cb) { _close_cb = cb; }
    void SetAnyEventCallback(const AnyEventCallback &cb) { _anyevent_cb = cb; }
    void EnableInactiveRelease(int timeout)
    {
        _timeout = timeout;
        _enable_inactive_release = true;
    }
    // 用于添加定时任务
    void RunAfter(const Functor &task, int delay)
    {
        _base_loop.RunInLoop(std::bind(&TcpServer::RunAfterInLoop, this, task, delay));
    }
    void Start()
    {
        _pool.Create();
        _base_loop.Start();
    }
};

class NetWork
{
public:
    NetWork()
    {
        lg(Info, "NetWork init.");
        signal(SIGPIPE, SIG_IGN);
    }
};

NetWork nw;